/********************************************************/
/*	Copyright (C) 2016 Gong Li Bin			*/
/*	Project:	GlbCloud			*/
/*	Author:		gong_libin			*/
/*	Date:		2012_01_10			*/
/*	File:		MrQueue.cpp			*/
/********************************************************/

#include "MrQueue.h"

CMrQueue::CMrQueue()
{
	m_ulTotal = 0;
	m_pHead = NULL;
	m_pTail = NULL;
	m_bMulti = true;
	if (MR_SUCCESS != pthread_cond_init(&m_stCond, NULL)) throw(strerror(errno));
	if (MR_SUCCESS != pthread_mutex_init(&m_stLock, NULL)) throw(strerror(errno));
}

CMrQueue::CMrQueue(bool bMulti)
{
	m_ulTotal = 0;
	m_pHead = NULL;
	m_pTail = NULL;
	m_bMulti = bMulti;
	if (true == bMulti) {
		if (MR_SUCCESS != pthread_cond_init(&m_stCond, NULL)) throw(strerror(errno));
		if (MR_SUCCESS != pthread_mutex_init(&m_stLock, NULL)) throw(strerror(errno));
	}
	else {
		memset(&m_stCond, '\0', sizeof(pthread_cond_t));
		memset(&m_stLock, '\0', sizeof(pthread_mutex_t));
	}
}

CMrQueue::~CMrQueue()
{
	void* pCur = NULL;

	m_ulTotal = 0;
	while (NULL != (pCur = m_pHead)) {
		m_pHead = *(void**)((UCHAR*)pCur + sizeof(UINT) + *(UINT*)pCur);
		free(pCur);
	}
	m_pTail = NULL;

	if (true == m_bMulti) {
		pthread_cond_destroy(&m_stCond);
		pthread_mutex_destroy(&m_stLock);
	}
}

void* CMrQueue::MrQueueHasNext(void* pCur)
{
	void* pReturn = NULL;

	if (true == m_bMulti) {
		MrQueueLock();
		while (0 == m_ulTotal) {
			pthread_cond_wait(&m_stCond, &m_stLock);
		}
	}

	if (NULL != pCur) {
		pReturn = *(void**)((UCHAR*)pCur + sizeof(UINT) + *(UINT*)pCur);
	}
	else {
		pReturn = m_pHead;
	}

	if (true == m_bMulti) {
		MrQueueUnlock();
	}

	return pReturn;
}

ULONG CMrQueue::MrQueueGetTotal()
{
	return m_ulTotal;
}

int CMrQueue::MrQueuePop(void *pData, UINT uiSize)
{
	void* pCur = NULL;
	int iReturn = MR_SUCCESS;

	if (true == m_bMulti) {
		MrQueueLock();
		while (0 == m_ulTotal) {
			pthread_cond_wait(&m_stCond, &m_stLock);
		}
	}

	pCur = m_pHead;
	if (NULL != pCur) {
		if (*(UINT*)pCur < uiSize) {
			(int)(m_ulTotal - 1) < 0 ? m_ulTotal = 0 : m_ulTotal -= 1;
			m_pHead = *(void**)((UCHAR*)pCur + sizeof(UINT) + *(UINT*)pCur);
			memcpy(pData, (UCHAR*)pCur + sizeof(UINT), *(UINT*)pCur);
			free(pCur);
		}
		else {
			iReturn = MR_FAILURE;
		}
	}
	else {
		iReturn = MR_FAILURE;
		m_ulTotal = 0;
	}

	if (true == m_bMulti) {
		MrQueueUnlock();
	}

	return iReturn;
}

int CMrQueue::MrQueuePut(void *pData, UINT uiSize)
{
	void* pPut = NULL;
	int iReturn = MR_SUCCESS;

	if (NULL != (pPut = malloc(sizeof(UINT) + sizeof(UCHAR) * uiSize + sizeof(void*)))) {
		memcpy(pPut, &uiSize, sizeof(UINT));
		memcpy((UCHAR*)pPut + sizeof(UINT), pData, uiSize);
		memset((UCHAR*)pPut + sizeof(UINT) + uiSize, '\0', sizeof(void*));

		if (true == m_bMulti) {
			MrQueueLock();
		}

		if (m_ulTotal > 0) {
			if (m_ulTotal < MR_QUEUE) {
				if (NULL != m_pTail) {
					memcpy((UCHAR*)m_pTail + sizeof(UINT) + *(UINT*)m_pTail, &pPut, sizeof(void*));
					m_pTail = pPut;
					m_ulTotal += 1;
				}
				else {
					MR_ERROR("Total error.\n");
					iReturn = MR_FAILURE;
				}
			}
			else {
				MR_ERROR("Total full.\n");
				iReturn = MR_FAILURE;
			}
		}
		else {
			m_pHead = m_pTail = pPut;
			m_ulTotal += 1;
		}

		if (true == m_bMulti) {
			MrQueueUnlock();
			pthread_cond_signal(&m_stCond);
		}
	}
	else {
		MR_ERROR("%s\n", strerror(errno));
		iReturn = MR_FAILURE;
	}

	return iReturn;
}

int CMrQueue::MrQueueGet(void *pData, UINT uiSize)
{
	void* pCur = NULL;
	void* pLeft = NULL;
	void* pRight = NULL;
	int iReturn = MR_FAILURE;

	if (true == m_bMulti) {
		MrQueueLock();
		while (0 == m_ulTotal) {
			pthread_cond_wait(&m_stCond, &m_stLock);
		}
	}

	pCur = pLeft = m_pHead;
	while (NULL != pCur) {
		pRight = *(void**)((UCHAR*)pCur + sizeof(UINT) + *(UINT*)pCur);
		if (*(UINT*)pCur == uiSize) {
			(int)(m_ulTotal - 1) < 0 ? m_ulTotal = 0 : m_ulTotal -= 1;
			memcpy(pData, (UCHAR*)pCur + sizeof(UINT), *(UINT*)pCur);
			if (m_pHead != pCur) {
				memcpy((UCHAR*)pLeft + sizeof(UINT) + *(UINT*)pLeft, pRight, sizeof(void*));
			}
			else {
				m_pHead = pRight;
			}
			iReturn = MR_SUCCESS;
			free(pCur);
			break;
		}
		pLeft = pCur;
		pCur = pRight;
	}

	if (true == m_bMulti) {
		MrQueueUnlock();
	}

	return iReturn;
}

void CMrQueue::MrQueueLock()
{
	if (MR_SUCCESS != pthread_mutex_lock(&m_stLock)) throw(strerror(errno));

	return;
}

void CMrQueue::MrQueueUnlock()
{
	if (MR_SUCCESS != pthread_mutex_unlock(&m_stLock)) throw(strerror(errno));

	return;
}
